#pragma once
#include "buffer.hpp"
#include <mutex>
#include <thread>
#include <atomic>
#include <functional>
#include <condition_variable>
namespace logs
{

    enum class AsyncType
    {
        ASYNC_SAFE,  // 安全模式下不会进行扩容，缓冲区满了就会阻塞
        ASYNC_UNSAFE // 会进行扩容，不考虑资源问题，常用于测试
    };
    // 这个类要让生产者缓冲区、消费者缓冲区都可以使用
    // 基于双缓冲区的生产者消费者模型，线程安全、线程同步
    using Functor = std::function<void(Buffer &)>;
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        // 传入一个可调用对象（消费消息的具体任务）
        AsyncLooper(const Functor &callBack, AsyncType safe_mode = AsyncType::ASYNC_SAFE) : _stop(false),
                                                                                            _thread(std::thread(&AsyncLooper::Entry, this)),
                                                                                            _call_Back(callBack),
                                                                                            _looper_type(safe_mode)
        {
            // 这里不能使用线程分离
            // 执行线程需要依赖主线程的数据（回调函数），因此需要等待线程执行完毕，否则就会出现异常
            // _thread.detach();
        }
        ~AsyncLooper()
        {
            stop();
        }
        // 提供给生产者线程的生产接口，放入数据
        void push(const char *data, size_t len)
        {
            // 将数据写入到生产缓冲区
            // 两种模式：固定容量大小，自动扩容；
            std::unique_lock<std::mutex> lock(_mtx);
            // 容量已满，需要等待读取
            if (_looper_type == AsyncType::ASYNC_SAFE)
            {
                _cond_pro.wait(lock, [&]()
                               { return len <= _pro_buf.able_write_size(); });
            }
            _pro_buf.push(data, len);

            // 有数据了，就可以唤醒消费者，让消费者消费数据
            _cond_con.notify_one();
        }
        void stop()
        {
            _stop = true;
            _cond_con.notify_all();
            _thread.join();
        }

    private:
        // 线程入口函数（消费线程）
        void Entry()
        {
            while (1)
            {
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    if (_stop && _pro_buf.empty())
                        break;
                    // 判断生产者缓冲区是否为空，为空则不能交换；
                    // 消费者消费生产者缓冲区
                    _cond_con.wait(lock, [&]()
                                   { return _stop || !_pro_buf.empty(); });
                    // 生产者缓冲区被消费, 交换之后就可以唤醒生产者了
                    _con_buf.swap(_pro_buf);
                    if (_looper_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                // 开始消费消息
                _call_Back(_con_buf);
                // 重置缓冲区状态
                _con_buf.reset();
            }
        }

    private:
        Functor _call_Back;

    private:
        AsyncType _looper_type;
        std::atomic<bool> _stop;
        Buffer _con_buf;                   // 消费者缓冲区
        Buffer _pro_buf;                   // 生产者缓冲区
        std::mutex _mtx;                   // 保证线程安全
        std::condition_variable _cond_pro; // 生产者条件变量
        std::condition_variable _cond_con; // 消费者条件变量
        std::thread _thread;               // 工作线程（实现缓冲区的交换）
    };
}